package brokerFailureController;

import java.util.HashMap;
import java.util.Set;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TopicSubscriber;

import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.JMSTextMessage;

import qpidClient.QpidClient;

/**
 * Receives messages from the detectors when they notice a change in the state
 * of a broker.
 * With that information, it deduces the state of every broker.
 * When a broker fails, it sends a message to the supercontroller informing
 * about it.
 */
public class Controller extends QpidClient implements MessageListener {
	
    //machine in which the controller is running
    protected final String CONTROLLER_IP;
    //topic used to communicate with the detector
    protected final String TOPIC_CONTROLLER = "brokerFailure.controller";
    //subscriber to listen to detectors
    protected TopicSubscriber subscriber;
    
    //topic used to communicate with the super controller
    protected final String TOPIC_SUPERCONTROLLER = "brokerFailure.superController";
    //publisher to send message to the super controller
    TopicPublisherAdapter scPublisher;
	
	//list of machines in which a broker is running
	protected static Set<String> brokers;
	
	//list of machines in which a broker is running and their state
    //state is an int from 0 (worst) to 10 (best), null if we don't know
    protected static HashMap<String, Integer> brokersState;
    
    //list of machines in which a broker is running and the states in which
    //they detect each one of the other brokers
    protected HashMap<String, HashMap<String, Integer>> brokersDetection;
    
    /**
     * Constructor.
     * 
     * @param controllerIP  IP address in which the controller broker is
     * 						running (its own IP address)
     * @param brokerAddresses  list of brokers in the system mapped to the
     * 						   ports they are listening at
     */
    public Controller(
    		String controllerIP, HashMap<String, Integer> brokerAddresses) {
    	
    	super();  //assign ID
    	
    	CONTROLLER_IP = controllerIP;
    	
    	brokers = brokerAddresses.keySet(); 
    	
    	//at the beggining, asume that all brokers are off
		brokersState = new HashMap<String, Integer>();
		for (String broker : brokers) {
			brokersState.put(broker, 0);
		}
		
		//at the beginning, asume that all detectors detect all the brokers off
		brokersDetection = new HashMap<String, HashMap<String, Integer>>();
		for (String broker : brokers) {
			HashMap<String, Integer> detection = new HashMap<String, Integer>();
			for (String brkr : brokers) {
				detection.put(brkr, 0);
			}
			brokersDetection.put(broker, detection);
		}
		
		//start a connection to read messages from the detectors
 	   	try {
 	   		
 	   		//create and start a connection and a session
			openConnAndSess(CONTROLLER_IP);

			//create a subscriber for listening to the detectors
			subscriber = getSubscriberForTopic(TOPIC_CONTROLLER);
			//set the subscriber to listen to messages
			subscriber.setMessageListener(this);
			
			System.out.println("Controller waiting for messages in "
							   + CONTROLLER_IP);

 	   	} catch (Exception e) {
			System.err.println(e.getMessage());
 	   	}
		
		while (true) {
			//wait to see if any detector reports any change in any broker
		}
		
    }
    
    /**
     * Compares the reported state of a broker with the current one and, if it
     * is necessary, informs the supercontroller about a broker failure.
     * 
     * @param host  which reported a change in a broker
     * @param broker  broker whose change is reported
     * @param state  new state of the broker who changed
     */
    public void reportChange(String host, String broker, int state) {
    	//save the detected state in the detected states of the broker
    	HashMap<String, Integer> states = brokersDetection.get(host);
		states.put(broker, state);
		brokersDetection.put(host, states);
		//get the current state of the broker in the controller
		int currentState = brokersState.get(broker);
		//if the reported state is not the current state
		if (state != currentState) {
			if (state == 0) { //failure
	    		brokersState.put(broker, state);
	    		System.out.println("The broker " + broker + " has failed");
	    		try {
					scPublisher = getPublisherForTopic(TOPIC_SUPERCONTROLLER);
					//create text to send
					String text = "broker failed:" + broker;
					//send the message to the super controller
					sendMessage(scPublisher, text);
				} catch (JMSException e) {
					e.printStackTrace();
				}
	    	} else if (state < currentState) { //worsening
	    		brokersState.put(broker, state);
	    		System.out.println("The state of the broker " + broker
	    						   + " is worsening to " + state);
	    	} else { //improvement to the worst state detected by any detector
	    			 //for a broker
	    		Set<String> detectors = brokersState.keySet();
	    		for (String detector : detectors) {
	    			int detectedState = brokersDetection.get(detector).get(broker);
	    			state = Math.min(state, detectedState);
	    		}
	    		if (state != currentState) {
					brokersState.put(broker, state);
					System.out.println("The state of the broker " + broker
									   + " is improving to " + state);
				}
	    	}
		}
		//if the reported state is the current state, do nothing
		System.out.println("State in the controller: " + brokersState);
    }

    /**
     * Read the detector, broker and state reported in a received message and
     * report the change.
     * 
     * This method is invoked asynchronously when a message is received.
     * 
     * @param msg  message received
     */
	@Override
	public void onMessage(Message msg) {
		try {
			JMSTextMessage message = (JMSTextMessage) msg;
			String text = message.getText();
			String[] parts = text.split(":");
			if (parts.length == 3) {
				reportChange(parts[0], parts[1], Integer.parseInt(parts[2]));
			} else {
				throw new Exception("Bad format of the message");
			}
		} catch (Exception e) {
			System.err.println("Error receiving or parsing message.");
		}		
	}

}
